package storage

import (
	"gim/comm"
	"encoding/gob"
	"fmt"
	"sync"
	"sync/atomic"
	"bytes"
	"strconv"
	"errors"
	"time"
	"container/list"
	log "github.com/golang/glog"
	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/util"
	"github.com/syndtr/goleveldb/leveldb/opt"
	"net/http"
	"net/rpc"
)

//1.存储离线消息到leveldb(只存储消息id)/已经确认的消息从这里删除
//2.存储客户端所有消息到leveldb(所有消息内容)
type Storage struct {
	host  string //ip:port
	root  string
	db_offline   *leveldb.DB //存储未确认的离线消息
	db_all *leveldb.DB //存储所有消息
	mutex sync.Mutex
}

type OfflineComparer struct{}

//{appid}_{uid}_{msgid}
func (oc OfflineComparer) Split(a []byte) ([]byte, []byte, []byte) {
	index1 := bytes.IndexByte(a, '_')
	if index1 == -1 || index1 + 1 >= len(a) {
		return nil, nil, nil
	}
	index2 := bytes.IndexByte(a[index1+1:], '_')
	if index2 == -1 || index2 + 1 >= len(a) {
		return nil, nil, nil
	}
	
	return a[:index1], a[index1+1:index1+1+index2], a[index1+1+index2+1:]
}

func (oc OfflineComparer) Compare(a, b []byte) int {
	p1, p2, p3 := oc.Split(a)
	p4, p5, p6 := oc.Split(b)

	if p1 == nil || p4 == nil {
		log.Infof("can't find seperate, a:%s b:%s compare bytes...\n", string(a), string(b))
		return bytes.Compare(a, b)
	}

	r1 := bytes.Compare(p1, p4)
	if r1 != 0 {
		return r1
	}

	r2 := bytes.Compare(p2, p5)
	if r2 != 0 {
		return r2
	}

	v1, err1 := strconv.ParseInt(string(p3), 10, 64)
	v2, err2 := strconv.ParseInt(string(p6), 10, 64)
	if err1 != nil || err2 != nil {
		log.Infof("parse int err, a:%s b:%s compare bytes...\n", string(a), string(b))
		return bytes.Compare(p3, p6)
	}
	
	if v1 < v2 {
		return -1
	} else if v1 == v2 {
		return 0
	} else {
		return 1
	}
}

func (OfflineComparer) Name() string {
	return "im.OfflineComparator"
}

func (OfflineComparer) Separator(dst, a, b []byte) []byte {
	return nil
}

func (OfflineComparer) Successor(dst, b []byte) []byte {

	return nil
}

func NewStorage(root string, host string) *Storage {
	storage := &Storage{
		root : root,
		host : host,
	}
	
	//初始化离线db
	opt := &opt.Options{Comparer:OfflineComparer{}}
	path := fmt.Sprintf("%s/%s", storage.root, "offlines")
	db_offline, err := leveldb.OpenFile(path, opt)
	if err != nil {
		log.Fatal("Open db offline failed", err)
	}
	
	storage.db_offline = db_offline
	
	//初始化消息库
	path = fmt.Sprintf("%s/%s", storage.root, "messages")
	db_all, err := leveldb.OpenFile(path, nil)
	if err != nil {
		log.Fatal("Open db all failed", err)
	}
	
	storage.db_all = db_all
	
	return storage
}

//获取消息接收着最大消息ID
func (storage *Storage) GetLastMessageID(appid int64, receiver int64) (int64, error) {
	key := fmt.Sprintf("%d_%d_0", appid, receiver)
	value, err := storage.db_all.Get([]byte(key), nil)
	if err != nil {
		log.Error("put err:", err)
		return 0, err
	}

	msgid, err := strconv.ParseInt(string(value), 10, 64)//10进制64位
	if err != nil {
		log.Error("parseint err:", err)
		return 0, err
	}
	return msgid, nil
}

//设置最近用户分配的最大消息ID
func (storage *Storage) SetLastMessageID(appid int64, receiver int64, msg_id int64) {
	key := fmt.Sprintf("%d_%d_0", appid, receiver)
	value := fmt.Sprintf("%d", msg_id)
	err := storage.db_all.Put([]byte(key), []byte(value), nil)
	if err != nil {
		log.Error("put err:", err)
		return
	}
}

//添加消息id到离线库
func (storage *Storage) AddOffline(msg_id int64, appid int64, receiver int64) bool {
	key := fmt.Sprintf("%d_%d_%d", appid, receiver, msg_id)
	value := fmt.Sprintf("%d", msg_id)
	log.Infof("%v", key)
	err := storage.db_offline.Put([]byte(key), []byte(value), nil)
	if err != nil {
		log.Error("put err:", err)
		return false
	}
	
	return true
}

//删除离线库消息
func (storage *Storage) RemoveOffline(msg_id int64, appid int64, receiver int64) {
	key := fmt.Sprintf("%d_%d_%d", appid, receiver, msg_id)
	err := storage.db_offline.Delete([]byte(key), nil)
	if err != nil {
		//can't get ErrNotFound
		log.Error("delete err:", err)
	}
}

//存储消息到消息库，包括消息内容
func (storage *Storage) AddAll(msg_id int64, appid int64, receiver int64, msg *comm.Message) bool {
	key := fmt.Sprintf("%d_%d_%d", comm.APP_ID, receiver, msg_id)
	//消息转化字节数组
	body := msg.ToData()
	buffer := new(bytes.Buffer)
	comm.WriteHeader(int32(len(body)), int64(msg.Seq), byte(msg.Cmd), byte(msg.Version), buffer)
	buffer.Write(body)
	value := buffer.Bytes()
	err := storage.db_all.Put([]byte(key), []byte(value), nil)
	if err != nil {
		log.Error("put err:", err)
		return false
	}
	
	return true
}

func (storage *Storage) ReadAll(msg_id int64, appid int64, receiver int64) *comm.Message {
	key := fmt.Sprintf("%d_%d_%d", comm.APP_ID, receiver, msg_id)
	
	value, err := storage.db_all.Get([]byte(key), nil)
	if err != nil {
		log.Error("storage db all get err:", err)
		return nil
	}
		
	_, seq, cmd, version := comm.ReadHeader(value)
	
	m := &comm.Message{
		Cmd : cmd,
		Seq : seq,
		Version : version,
	}
	
	im := &comm.IMMessage{}
	im.FromData(value[16:])
	
	m.Body = im
	return m
}

type StorageServer struct {
	storage *Storage
}
//rpc调用方法
//保存消息，返回含有消息id的消息
func (storageSever *StorageServer) SaveMessage(args *comm.Message, reply *int64) error {
	
	body := args.Body.(comm.IMMessage)
	
	storageSever.storage.mutex.Lock()
	//获取最大msgid
	msgid, err := storageSever.storage.GetLastMessageID(comm.APP_ID, body.Receiver)
	if err != nil && err != leveldb.ErrNotFound {
		log.Error("get max msg id err:", err)
	}
	
	//自增1
	atomic.AddInt64(&msgid, 1)
	
	//存储最大msgid
	storageSever.storage.SetLastMessageID(comm.APP_ID, body.Receiver, msgid)
	storageSever.storage.mutex.Unlock()
	
	m := comm.Message{
		Cmd : args.Cmd,
		Seq : args.Seq,
		Version: args.Version,
	}
	
	m.Body = &comm.IMMessage{
		Sender : body.Sender,
		Receiver : body.Receiver,
		Gid: body.Gid,
		Timestamp : body.Timestamp,
		Msgid : msgid,
		Content : body.Content,
		Extra : body.Extra,
	}
	
	//存储到消息库
	hasSave := storageSever.storage.AddAll(msgid, comm.APP_ID, body.Receiver, &m)
	if !hasSave {
		return errors.New("message save all err")
	}
	
	//存储到离线库
	hasSave = storageSever.storage.AddOffline(msgid, comm.APP_ID, body.Receiver)
	if !hasSave {
		return errors.New("message save offline err")
	}
	
	//返回完整消息
	*reply = msgid
	
	return nil
}

//rpc调用方法
//确认消息，删除离线消息库里面的消息
func (storageSever *StorageServer) AckMessage(args *comm.MessagePeerACK, reply *bool) error {
	storageSever.storage.RemoveOffline(args.Msgid, comm.APP_ID, args.Receiver)
	*reply = true
	return nil
}

//rpc调用方法
//获取离线消息
func (storageSever *StorageServer) GetMessages(args *comm.AppUserID, reply *[]comm.Message) error {
	
	appid := int64(comm.APP_ID)
	uid := args.Uid
	
	log.Infof("load offline message appid:%d uid:%d\n", appid, uid)
	c := make([]comm.Message, 0, 10)
	start := fmt.Sprintf("%d_%d_1", appid, uid)
	end := fmt.Sprintf("%d_%d_9223372036854775807", appid, uid)

	r := &util.Range{Start:[]byte(start), Limit:[]byte(end)}
	iter := storageSever.storage.db_offline.NewIterator(r, nil)
	for iter.Next() {
		value := iter.Value()
		msgid, err := strconv.ParseInt(string(value), 10, 64)
		if err != nil {
			log.Error("parseint err:", err)
			continue
		}
		log.Info("offline msgid:", msgid)
		
		m := storageSever.storage.ReadAll(msgid, appid, uid)
		
		c = append(c, *m)
	}
	iter.Release()
	err := iter.Error()
	if err != nil {
		log.Warning("iterator err:", err)
	}
	log.Info("offline count:", len(c))
	
	*reply = c
	
	return nil
}

func StartStorage(root string, host string){
	storage := NewStorage(root, host)
	storageServer := new(StorageServer)
	storageServer.storage = storage
	
	gob.Register(comm.IMMessage{})
	
	rpc.Register(storageServer)
	rpc.HandleHTTP()
	
	go func() {
		err := http.ListenAndServe(storage.host, nil)
		if err != nil {
			log.Error("start storage err:", err)
		}
	}()
}

type StorageConn struct {
	Conn *rpc.Client
	E    bool
}

func NewStorageConn() *StorageConn {
	c := new(StorageConn)
	return c
}

func (client *StorageConn) Dial(addr string) error {
	conn, err := rpc.DialHTTP("tcp", addr)
	if err != nil {
		client.E = true
		return err
	}
	client.Conn = conn
	return nil
}

func (client *StorageConn) Close() {
	if client.Conn != nil {
		client.Conn.Close()
	}
}

var nowFunc = time.Now // for testing

type idleConn struct {
	c *StorageConn
	t time.Time
}

type StorageConnPool struct {

	Dial           func()(*StorageConn, error)

	// Maximum number of idle connections in the pool.
	MaxIdle int

	// Maximum number of connections allocated by the pool at a given time.
	// When zero, there is no limit on the number of connections in the pool.
	MaxActive int

	// Close connections after remaining idle for this duration. If the value
	// is zero, then idle connections are not closed. Applications should set
	// the timeout to a value less than the server's timeout.
	IdleTimeout time.Duration

	// mu protects fields defined below.
	mu     sync.Mutex
	closed bool
	active int

	// Stack of idleConn with most recently used at the front.
	idle list.List
	
	sem  chan int
}

func NewStorageConnPool(max_idle int, max_active int, 
	idle_timeout time.Duration, 
	dial func()(*StorageConn, error)) *StorageConnPool {
	if max_idle > max_active {
		return nil
	}
	pool := new(StorageConnPool)
	pool.MaxIdle = max_idle
	pool.MaxActive = max_active
	pool.IdleTimeout = idle_timeout
	pool.Dial = dial

	pool.sem = make(chan int, max_active)
	for i := 0; i < max_active; i++ {
		pool.sem <- 0
	}
	return pool
}

func (p *StorageConnPool) Get() (*StorageConn, error) {
	<- p.sem

	p.mu.Lock()
	// Prune stale connections.
	if timeout := p.IdleTimeout; timeout > 0 {
		for i, n := 0, p.idle.Len(); i < n; i++ {
			e := p.idle.Back()
			if e == nil {
				break
			}
			ic := e.Value.(idleConn)
			if ic.t.Add(timeout).After(nowFunc()) {
				break
			}
			p.idle.Remove(e)
			p.active -= 1
			p.mu.Unlock()
			ic.c.Close()
			p.mu.Lock()
		}
	}

	// Get idle connection.
	for i, n := 0, p.idle.Len(); i < n; i++ {
		e := p.idle.Front()
		if e == nil {
			break
		}
		ic := e.Value.(idleConn)
		p.idle.Remove(e)
		p.mu.Unlock()
		return ic.c, nil
	}

	if p.MaxActive > 0 && p.active >= p.MaxActive {
		log.Error("storage pool exhausted")
		p.sem <- 0
		return nil, errors.New("exhausted")
	}

	// No idle connection, create new.
	dial := p.Dial
	p.active += 1
	p.mu.Unlock()
	c, err := dial()
	if err != nil {
		p.mu.Lock()
		p.active -= 1
		p.mu.Unlock()
		c = nil
		p.sem <- 0
	}
	return c, err
}

func (p *StorageConnPool) Release(c *StorageConn) {
	defer func() {
		p.sem <- 0
	}()

	if !c.E {
		p.mu.Lock()
		p.idle.PushFront(idleConn{t: nowFunc(), c: c})
		if p.idle.Len() > p.MaxIdle {
			c = p.idle.Remove(p.idle.Back()).(idleConn).c
		} else {
			c = nil
		}
		p.mu.Unlock()
	}
	if c != nil {
		p.mu.Lock()
		p.active -= 1
		p.mu.Unlock()
		c.Close()
	}
}

func DialStorageFun(addr string) func()(*StorageConn, error) {
	f := func() (*StorageConn, error){
		storage := NewStorageConn()
		err := storage.Dial(addr)
		if err != nil {
			log.Error("connect storage err:", err)
			return nil, err
		}
		return storage, nil
	}
	return f
}